// @Author EthanScriptOn
// @Desc
package rocketmq_operator

import (
	"context"
	"fmt"
	"gitee.com/fatzeng/srf_switch_basic_components/common/identity"
	"gitee.com/fatzeng/srf_switch_expansion_package/connect"
	"gitee.com/fatzeng/srf_switch_expansion_package/extend_config"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/actuator"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/operator"
	"github.com/apache/rocketmq-client-go/v2/primitive"
)

type RocketmqProduceOperator struct {
	*identity.IdentityInformationCarrier
}

func (r *RocketmqProduceOperator) Execute(ctx *actuator.SrfSwitchBridgeActuatorContext, configIdentifier interface{}, structureParam interface{}) (interface{}, error) {
	rocketmqProducer, err := connect.GetRocketmqProducer()
	if err != nil {
		return nil, err
	}

	messages, ok := structureParam.([]*primitive.Message)
	if !ok {
		return nil, fmt.Errorf("structureParam is not []*primitive.Message")
	}
	if len(messages) <= 0 {
		return nil, fmt.Errorf("messages is empty")
	}
	rocketmqOperatorConfig := operator.GetOperatorConfig(configIdentifier)
	if rocketmqOperatorConfig == nil {
		return nil, fmt.Errorf("rocketmqOperatorConfig is nil")
	}
	config, ok := rocketmqOperatorConfig.(*extend_config.RocketmqProduceOperatorConfig)
	if !ok {
		return nil, fmt.Errorf("rocketmqOperatorConfig is not *RocketmqProduceOperatorConfig")
	}
	timeout, _ := context.WithTimeout(context.Background(), config.TimeOut)

	responses := make([]*extend_config.RocketmqOperatorResponse, 0)
	for _, message := range messages {
		res, err := rocketmqProducer.SendSync(timeout, message)
		responses = append(responses, &extend_config.RocketmqOperatorResponse{Response: res, Err: err, OriginMsg: message})
	}
	return responses, nil
}
